package com.hopu.shop;

import com.alibaba.fastjson.JSONObject;

import com.hopu.bean.shop.Top10HotShop;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class Top10HotShopAna {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("top10HotShopAna").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        JavaPairRDD<Top10HotShop, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String sid = json.getString("shop_id");
            String sname = json.getString("shop_name");
            String type = json.getString("event_type");

            return new Tuple2<>(new Top10HotShop(sid, sname, type), 1);
        });
        JavaPairRDD<Top10HotShop, Integer> viewRDD = pairRDD.filter(t -> "view".equals(t._1.getType()));
        JavaPairRDD<Top10HotShop, Integer> cartRDD = pairRDD.filter(t -> "cart".equals(t._1.getType()));
        JavaPairRDD<Top10HotShop, Integer> parchaseRDD = pairRDD.filter(t -> "parchase".equals(t._1.getType()));

        //根据Sid进行分组统计
        viewRDD = groupBySid(viewRDD);
        cartRDD = groupBySid(cartRDD);
        parchaseRDD = groupBySid(parchaseRDD);


        JavaPairRDD<Top10HotShop, Tuple2<Tuple2<Integer, Optional<Integer>>, Optional<Integer>>> result = viewRDD.leftOuterJoin(cartRDD).leftOuterJoin(parchaseRDD);

        JavaPairRDD<Top10HotShop, Integer> top10Hot = result.mapToPair(t -> {
            Top10HotShop top10HotShop = t._1;
            int view = t._2._1._1;
            int cart = t._2._1._2.get();
            int parchase = t._2._2.get();
            top10HotShop.setCart(cart);
            top10HotShop.setParchase(parchase);
            return new Tuple2<>(top10HotShop, view + cart + parchase);
        });

        //排序取出前十
        List<Top10HotShop> top10HotShops = top10Hot.sortByKey().map(t -> t._1).take(10);
        top10HotShops.forEach(t -> System.out.println(t));

        //写入到数据库
        String[] columns = {"sid", "sname", "view", "cart", "parchase"};
        for (Top10HotShop top10HotShop: top10HotShops) {
            String[] values = {top10HotShop.getSid(), top10HotShop.getSname(),
            top10HotShop.getView()+"", top10HotShop.getCart()+"", top10HotShop.getParchase()+""};

            try {
                HBaseUtils.putOneRowToHbase("shopana", top10HotShop.getSid(), "top10hot", columns, values);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }


    }

    public static JavaPairRDD<Top10HotShop, Integer> groupBySid(JavaPairRDD<Top10HotShop, Integer> rdd) {
        JavaPairRDD<Top10HotShop, Iterable<Integer>> group = rdd.groupByKey();

        JavaPairRDD<Top10HotShop, Integer> pairRDD = group.mapToPair(t -> {
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            if (t._1.getType().equals("view")) {
                t._1.setView(count);
            } else if (t._1.getType().equals("cart")) {
                t._1.setCart(count);
            } else {
                t._1.setParchase(count);
            }
            return new Tuple2<>(t._1, count);
        });

        return pairRDD;
    }



}
